'''RDD数据计算练习案例2
需求,复制以上内容到文件中，使用Spark读取文件进行计算:
各个城市销售额排名，从大到小？
全部城市，有哪些商品类别在售卖？
北京市有哪些商品类别在售卖？
'''

from pyspark import SparkConf, SparkContext
import json
# 使用os 配置python安装位置， 帮助spark找到python;
import os
os.environ['PYSPARK_PYTHON'] = 'D:\\yfxdeve\\python\Python39\\python.exe'

conf = SparkConf().setMaster("local[*]").setAppName("test_park")
sc = SparkContext(conf=conf)


# todo 需求1：各个城市销售额排名，从大到小？
# 1.1 读取指定文件
rdd1 = sc.textFile("D:\\hxy\\test\\练习案例数据.txt")
# 1.2 根据 | 进行切割
rdd2 = rdd1.flatMap(lambda x: x.split("|"))
# 1.3 将数组中的json转为字典dict
rdd3 = rdd2.map(lambda x: json.loads(x))
print(rdd3.collect())
# 1.4 获取 城市areaName 与 销售额money 由于金额是字符串这里转为int;
rdd4 = rdd3.map(lambda x: (x['areaName'], int(x['money'])))
# 1.5 根据城市分组并统计销售额
rdd5 = rdd4.reduceByKey(lambda a, b: a+b)
# 1.6 根据销售额排序 从大到小
rdd6 = rdd5.sortBy(lambda x:x[1], ascending=False, numPartitions=1)
print(rdd6.collect()) #最终结果： [('北京', 39713), ('上海', 23608), ('杭州', 19204), ('苏州', 10840)]

# todo 需求2：全部城市，有哪些商品类别在售卖？
# 2.1 获取城市与商品类别，并去重相同城市下同一商品类别；
test2RDD1 = rdd3.map(lambda x: (x['areaName'], x['category'])).distinct()
test2RDD2 = test2RDD1.reduceByKey(lambda a, b: a+'_'+b)
print(test2RDD2.collect()) #最终结果：[('杭州', '书籍_电脑_家具_手机'), ('苏州', '电脑_手机_家具'), ('北京', '手机_电脑_家具_平板电脑_家电'), ('上海', '电脑_家具_手机')]

# todo 需求3：北京市有哪些商品类别在售卖？
# 3.1 指过滤出北京的商品类别
test3RDD1 = rdd3.filter(lambda x: x['areaName'] == '北京')
# 3.2 对商品类别去重
test3RDD2 = test3RDD1.map(lambda x: x['category']).distinct()
print(test3RDD2.collect()) #最终结果：['平板电脑', '家电', '手机', '电脑', '家具']


# 案例结束

# 关闭
sc.stop()

